package logmonitor;

import backtype.storm.Config;
import backtype.storm.StormSubmitter;
import backtype.storm.topology.TopologyBuilder;
import logmonitor.bolt.FilterBolt;
import logmonitor.bolt.PrepareRecordBolt;
import logmonitor.bolt.SaveMessageBolt;
import storm.kafka.KafkaSpout;
import storm.kafka.SpoutConfig;
import storm.kafka.ZkHosts;

/**
 * author:CN.CDG
 * Date:2019/2/23
 * Time:14:44
 * 驱动类
 **/
public class LogMonitorTopplogy {
    public static void main(String[] args) throws Exception{
        //创建一个拓扑构建类
        final TopologyBuilder builder = new TopologyBuilder();
        //指定kafka的topic
        String topic="logmonitor";
        //指定kafka的集群列表
        final ZkHosts zkHosts = new ZkHosts("hadoop01:2181");
        //初始化kafka的配置信息
        final SpoutConfig spoutConfig = new SpoutConfig(zkHosts, topic, "/opt/app/stormdir/storm", "Logmonitor");
        //调用builder来指定自定义spout
        builder.setSpout("kafkaspout",new KafkaSpout(spoutConfig),1);
        //调用builder来自定义bolt，用来校验数据
        builder.setBolt("filterbolt",new FilterBolt(),2).shuffleGrouping("kafkaspout");
        //调用builder来指定自定义bolt，用来把触发信息发送给负责人
        builder.setBolt("preparerecordbolt",new PrepareRecordBolt(),1).shuffleGrouping("filterbolt");
        //调用builder来指定自定义bolt来存储触发信息
        builder.setBolt("savemessagebolt",new SaveMessageBolt(),1).shuffleGrouping("preparerecordbolt");
        //创建配置信息类
        final Config conf = new Config();
        //配置是否生成运行日志
        conf.setDebug(false);//不生成
        //配置运行环境（本地还是在jiqun）
        if(args!=null&&args.length>0){
            //集群模式，就要配置启动几个worker
            conf.setNumWorkers(3);
            //向集群提交任务
            StormSubmitter.submitTopologyWithProgressBar(args[0],conf,builder.createTopology());
        }else{
            //是在本地模式下，需要设置线程数来模拟集群环境
            conf.setMaxTaskParallelism(3);
            //提交任务
            StormSubmitter.submitTopologyWithProgressBar(args[0],conf,builder.createTopology());
        }
    }
}
